package com.tyssq.rocketmq.consumer;

import com.tyssq.rocketmq.pojo.exception.ReconsumeLaterException;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;

/**
 * <p>
 * 消费者
 * </p>
 *
 * @author liaozhenghan
 * @date 2021-06-02 15:00:54
 */
@Slf4j
@Getter
public abstract class BaseConsumer extends DefaultMQPushConsumer implements InitializingBean {
    private String consumerGroupName;
    private String topicName;
    @Autowired
    private Environment env;

    @Override
    public void afterPropertiesSet() throws Exception {

        // 从配置文件获取分组和主题的开头配置
        String prefix = env.getProperty("rocketmq.prefix");
        if (prefix == null) {
            prefix = "";
        }

        // 获取分组和主题
        String groupName = this.getGroupName();
        String topic = this.getTopic();
        if (groupName == null) {
            log.error("{}：未配置分组", this.getClass().getSimpleName());
            throw new Exception(this.getClass().getSimpleName() + "：未配置分组");
        }
        if (topic == null) {
            log.error("{}：未配置主题", this.getClass().getSimpleName());
            throw new Exception(this.getClass().getSimpleName() + "：未配置主题");
        }

        this.consumerGroupName = prefix + this.getGroupName();
        this.topicName = prefix + this.getTopic();

        log.info("消费者：{}，主题：{}，分组：{}，正在创建---------------------------------------",
                this.getClass().getSimpleName(), topicName, consumerGroupName);
        this.setConsumerGroup(consumerGroupName);
        this.subscribe(topicName, getSubExpression());
        this.initMessageListener();
        this.beforeStart();
        this.start();
        log.info("消费者：{}，主题：{}，分组：{}，创建成功---------------------------------------",
                this.getClass().getSimpleName(), topicName, consumerGroupName);
    }

    @Override
    @Value("${rocketmq.namesrvAddr}")
    public void setNamesrvAddr(String namesrvAddr) {
        super.setNamesrvAddr(namesrvAddr);
    }

    protected String getGroupName() {
        return getTopic();
    }

    protected abstract String getTopic();

    protected String getSubExpression() {
        return "*";
    }

    protected void beforeStart() {
    }

    protected void initMessageListener() {
        this.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() > 1) {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            for (MessageExt msg : msgs) {
                try {
                    long start = System.currentTimeMillis();
                    dealMessage(msg);
                    long end = System.currentTimeMillis();
                    log.info("主题：{}，类：{}，消息ID：{}，keys：{}，消费耗时：{}毫秒", msg.getTopic(),
                            this.getClass().getSimpleName(), msg.getMsgId(), msg.getKeys(), end - start);
                } catch (ReconsumeLaterException e) {
                    log.info("延迟消费！keys：{}，原因：{}", msg.getKeys(), e.getMessage());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                } catch (Exception e) {
                    log.error("消息消费失败！keys：{}，失败信息：{}", msg, e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }

    protected  void dealMessage(MessageExt msg) {
        dealMessage(msg.getMsgId(), new String(msg.getBody()));
    }


    /**
     * <p>
     * 消息消费
     * </p>
     *
     * @param messageId 消息ID
     * @param message   消息
     */
    protected abstract void dealMessage(String messageId, String message);

}
